package MapReduce.Demo12_BookJoin.mapJoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import utils.JobSubmit;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author:lixinlei
 * @Date:2022/3/24 11:10
 **/
public class BookMapJoinApp {

    public static class BMJMapper extends Mapper<LongWritable, Text,Text, NullWritable>{

        Map<String,String> bookMap = new HashMap<String,String>();
        Map<String,String> personMap = new HashMap<String,String>();

        Text outKey = new Text();
        NullWritable outValue = NullWritable.get();

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {

            //获取图书和人员文件的路径
            URI[] uris = context.getCacheFiles();
            String bookFile = uris[0].getPath();
            String personFile = uris[1].getPath();

            //分别通过路径创建输入流
            BufferedReader br_book = new BufferedReader(new FileReader(bookFile));
            BufferedReader br_person = new BufferedReader(new FileReader(personFile));

            String bookInfo = "";
            while((bookInfo = br_book.readLine()) !=null){
                String[] fields_book = bookInfo.split("\t");
                bookMap.put(fields_book[0],fields_book[1]);
            }

            String personInfo = "";
            while((personInfo = br_person.readLine()) !=null){
                String[] fields_person = personInfo.split("\t");
                personMap.put(fields_person[0],fields_person[1]);
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t");

            String bookId = fields[0];
            String personId = fields[1];
            String dateTime = fields[2];

            String singleData = bookId+"\t"+bookMap.get(bookId)+"\t"+
                    personId+"\t"+personMap.get(personId)+"\t"+dateTime;

            outKey.set(singleData);

            context.write(outKey,outValue);

        }
    }

    public static void main(String[] args) throws URISyntaxException {

        URI[] uris = new URI[2];
        uris[0] = new URI("file:///E:/data_hadoop/BookJoin/book/book1.txt");
        uris[1] = new URI("file:///E://data_hadoop/BookJoin/person/person.txt");

        JobSubmit.submitBaseJob(
                BookMapJoinApp.class,
                args,
                0,
                uris
        );
    }

}
